package jobworker

import (
	"fmt"
	"gitee.com/captials-team/ubdframe/src/common/utils"
	"gitee.com/captials-team/ubdframe/src/domain/configstc"
	v1log "gitee.com/captials-team/ubdframe/src/pkg/logs"
	"gitee.com/captials-team/ubdframe/src/pkg/metrics"
	"go.uber.org/dig"
	"strings"
	"sync"
	"time"
)

type WorkerMgr struct {
	di  *dig.Container
	cfg *configstc.JobConfig
	l   v1log.ILog

	//运行的worker
	workers []IWorker
	//注册的可用worker
	regWorkers map[string]YieldWorkerFunc

	mu      sync.RWMutex
	monitor chan metrics.IndicData

	indicators []metrics.IIndicator //上报指标
}

func (mgr *WorkerMgr) Name() string {
	return "worker_mgr"
}

func NewWorkerMgr(di *dig.Container, c *configstc.JobConfig, l v1log.ILog) *WorkerMgr {
	mgr := &WorkerMgr{
		di:         di,
		cfg:        c,
		l:          l,
		regWorkers: map[string]YieldWorkerFunc{},
		monitor:    make(chan metrics.IndicData, 64),
	}

	return mgr
}

func (mgr *WorkerMgr) Start() error {
	mgr.loadFiles()
	mgr.loadWorkers()

	go mgr.runMonitor(mgr.monitor) //运行监控
	mgr.runWorkers(mgr.workers)    //运行workers
	return nil
}

func (mgr *WorkerMgr) RunOnce(times ...int64) {
	mgr.loadFiles()
	mgr.loadWorkers()

	go mgr.runMonitor(mgr.monitor)        //运行监控
	mgr.runWorkers(mgr.workers, times...) //运行workers
}

func (mgr *WorkerMgr) Stop() error {
	mgr.stopWorkers(mgr.workers)
	return nil
}

// StartAction 单独启动一个action处理
func (mgr *WorkerMgr) StartAction(item configstc.MultiActionItem) error {
	workers := mgr.actionItem2Workers(0, item)

	mgr.runWorkers(workers)
	return nil
}

// AsyncStartAction 异步-StartAction
func (mgr *WorkerMgr) AsyncStartAction(item configstc.MultiActionItem) {
	go mgr.StartAction(item)
}

// runWorkers 启动worker
func (mgr *WorkerMgr) runWorkers(workers []IWorker, times ...int64) {
	mgr.l.Info("start %d workers", len(workers))
	if len(workers) == 0 {
		return
	}
	//只运行指定次数
	count := utils.MoreValueChooseInt64(0, times...)

	var wg sync.WaitGroup

	for _, w := range workers {
		mgr.l.Info("worker start: %s, id=%s", w.Name(), w.Op().OpId)
		mgr.l.Info("worker Op config[%s-%s] %+v", w.Name(), w.Op().OpId, w.Op())
		copyW := w

		wg.Add(1)
		go func() {
			defer wg.Done()

			if count > 0 {
				copyW.Run(count)
			} else {
				copyW.Start()
			}
		}()
	}

	mgr.l.Ctl(len(workers) == 0).Warn("workers empty")
	wg.Wait()
}

func (mgr *WorkerMgr) stopWorkers(workers []IWorker) {
	for _, v := range workers {
		v.Stop()
	}
}

func (mgr *WorkerMgr) runMonitor(ch chan metrics.IndicData) {
	defer mgr.l.Info("worker monitor stop")

	for {
		select {
		case data, ok := <-ch:
			if !ok {
				return
			}
			//mgr.l.Info("get monitor data")
			for _, v := range mgr.indicators {
				v.Report(data)
			}
		}
	}
}

// AsyncStart 异步启动
func (mgr *WorkerMgr) AsyncStart() {
	go mgr.Start()
}

func (mgr *WorkerMgr) loadFiles() {
	for _, f := range mgr.cfg.MultiFiles {
		var c configstc.JobConfig
		err := utils.ReadYamlFile(&c, f)
		if err != nil {
			panic(fmt.Sprintf("read config yaml: %s", err))
		}
		mgr.l.Info("load worker file %s", f)
		//每个文件内的globalParams独立于全局的globalParams
		for k, v := range c.MultiActions {
			c.MultiActions[k].ExtendParams = mgr.mergeMap(mgr.cfg.GlobalParams, c.GlobalParams, v.ExtendParams)
		}

		mgr.cfg.MultiActions = append(mgr.cfg.MultiActions, c.MultiActions...)
		//mgr.cfg.GlobalParams = mgr.mergeMap(mgr.cfg.GlobalParams, c.GlobalParams)
	}
}

func (mgr *WorkerMgr) loadWorkers() {
	var workers []IWorker

	for i, item := range mgr.cfg.MultiActions {
		workers = append(workers, mgr.actionItem2Workers(i, item)...)
	}
	mgr.l.Info("use workers num: %d", len(workers))
	mgr.workers = workers
}

func (mgr *WorkerMgr) actionItem2Workers(index int, item configstc.MultiActionItem) []IWorker {
	var workers []IWorker
	item.Id = index
	//将配置的动作(Action)转为实际操作参数（JobParam）
	ops := mgr.action2JobParams(mgr.cfg, item)
	if len(ops) <= 0 {
		mgr.l.Warn("ops is empty %+v", item)
		return workers
	}
	mgr.l.Info("ops: %+v", ops)

	//对每个op生成worker
	for _, op := range ops {
		w, err := mgr.newWorker(op)
		if err != nil {
			mgr.l.Panic("workerErr %s", err)
			continue
		}
		//worker支持monitor
		if m, ok := w.(IWorkerMonitor); ok {
			m.SetMonitor(mgr.monitor)
		}
		workers = append(workers, w)
	}
	return workers
}

// Reg 注册worker
func (mgr *WorkerMgr) Reg(f YieldWorkerFunc, moreNames ...string) {
	if f == nil {
		return
	}
	moreNames = append(moreNames, f(mgr.di, nil).Actions()...)

	for _, s := range moreNames {
		//已注册的不处理
		if _, exist := mgr.regWorkers[s]; exist {
			continue
		}

		mgr.l.Info("worker registered => %s", s)
		mgr.regWorkers[s] = f
	}
}

// RegIndicator 注册指标收集
func (mgr *WorkerMgr) RegIndicator(indicator metrics.IIndicator) {
	mgr.indicators = append(mgr.indicators, indicator)
}

// RegAction 注册动作（动作自动转换为worker）
//
//	注意动作设置至少一个moreNames，不设置会成为无名worker，将匹配不到op
func (mgr *WorkerMgr) RegAction(f DoWorkerActionFunc, moreNames ...string) {
	mgr.Reg(f.YieldWorkerFunc, moreNames...)
}

func (mgr *WorkerMgr) action2JobParams(cfg *configstc.JobConfig, item configstc.MultiActionItem) []*JobParam {
	//生成worker并append
	count := item.Count
	if count <= 0 {
		count = 1
	}

	actions := strings.Split(item.Actions, ",")
	extendParams := mgr.mergeMap(cfg.GlobalParams, item.ExtendParams)

	var ops []*JobParam
	index := 0
	for _, act := range actions {
		if len(act) <= 0 {
			continue
		}

		//指定count进行repeat创建
		for i := 0; i < count; i++ {
			ops = append(ops, &JobParam{
				OpId:         mgr.itemOpId(item, index),
				Action:       act,
				ExtendParams: extendParams,
				Interval:     time.Millisecond * time.Duration(item.OpInterval),

				Statistical: true, //默认启用
			})
			index++
		}
	}
	return ops
}

func (mgr *WorkerMgr) itemOpId(item configstc.MultiActionItem, index int) string {
	count := item.Count
	if count <= 0 {
		count = 1
	}
	actions := strings.Split(item.Actions, ",")
	if len(actions) > 1 || count > 1 {
		return fmt.Sprintf("worker_%d_%d", item.Id, index)
	}
	return fmt.Sprintf("worker_%d", item.Id)
}

func (mgr *WorkerMgr) newWorker(op *JobParam) (IWorker, error) {
	fc, exist := mgr.regWorkers[op.Action]
	if !exist {
		return nil, fmt.Errorf("action=%s not match worker", op.Action)
	}

	w := fc(mgr.di, op)

	//params不为nil则受限于提供的params
	if param := w.Params(); param != nil {
		op.ExtendParams = mgr.coverMap(param, op.ExtendParams)
	}

	return w, nil
}

// mergeExtendParams 合并
//
//	mParams里的值覆盖gParams的
func (mgr *WorkerMgr) mergeMap(gParams map[string]string, mParamsArr ...map[string]string) map[string]string {
	params := map[string]string{}
	if gParams != nil {
		for k, v := range gParams {
			params[k] = v
		}
	}
	for _, mParams := range mParamsArr {
		if mParams != nil {
			for k, v := range mParams {
				params[k] = v
			}
		}
	}
	return params
}

// coverMap 覆盖map，以m1为准，m2只覆盖m1有的key
func (mgr *WorkerMgr) coverMap(m1 map[string]string, m2 map[string]string) map[string]string {
	m := map[string]string{}
	if m1 != nil {
		for k, v := range m1 {
			m[k] = v
		}
	}
	if m2 != nil {
		for k := range m2 {
			if _, exist := m[k]; exist {
				m[k] = m2[k]
			}
		}
	}
	return m
}
